﻿using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
using RabbitMQ.Client;
using RabbitMQ.Client.Content;
using System.Collections.Concurrent;
using RabbitMQ.Client.Events;

namespace Keede.RabbitMQ
{
    public class Client : BaseClient
    {
        IBasicProperties _basicProperties;
        readonly ConcurrentDictionary<string, IQueueListener> _queueListener;
        readonly bool _isLogDetail;

        /// <summary>
        /// client
        /// </summary>
        /// <param name="auth">auth</param>
        /// <param name="isLogDetails">Is need log deatail message,for example each one publish message</param>
        /// <param name="prefetchCount">Pre fetch count</param>
        public Client(Authorization auth, bool isLogDetails = false, ushort prefetchCount = 20) : base(auth, prefetchCount)
        {
            _queueListener = new ConcurrentDictionary<string, IQueueListener>();
            _isLogDetail = isLogDetails;
        }

        /// <summary>
        /// queue declare
        /// </summary>
        public void QueueDeclare(string queue, bool durable = true, bool exclusive = false, bool autoDelete = false, IDictionary<string, object> arguments = null)
        {
            _channel.QueueDeclare(queue, durable, exclusive, autoDelete, arguments);
        }

        /// <summary>
        /// queue bind
        /// </summary>
        public void QueueBind(string exchange, string queue, string routingKey)
        {
            _channel.QueueBind(queue, exchange, routingKey);
        }

        /// <summary>
        /// exchange declare
        /// </summary>
        public void ExchangeDeclare(string exchange, ExchangeType exchangeType, bool durable = true, bool autoDelete = false, IDictionary<string, object> arguments = null)
        {
            _channel.ExchangeDeclare(exchange, exchangeType.ToString().ToLower(), durable, autoDelete, arguments);
        }


        /// <summary>
        /// get base properties
        /// </summary>
        /// <returns></returns>
        public IBasicProperties GetBasicProperties()
        {
            var mapMessageBuilder = new MapMessageBuilder(this._channel);
            return (IBasicProperties)mapMessageBuilder.GetContentHeader();
        }

        /// <summary>
        /// publish message
        /// </summary>
        /// <param name="exchange">exchange</param>
        /// <param name="routingKey">routing key</param>
        /// <param name="message">message</param>
        /// <param name="basicProperties">basicProperties</param>
        public void Publish(string message, string exchange = "", string routingKey = "", IBasicProperties basicProperties = null)
        {
            if (string.IsNullOrEmpty(message)) throw new ArgumentNullException(nameof(message));

            lock (this)
            {
                if (_closed) throw new Exception("Rabbitmq connection already closed!");
                var body = Encoding.UTF8.GetBytes(message);

                this._channel.BasicPublish(exchange, routingKey, basicProperties, body);
                if (_isLogDetail) _log.Info($"[publish] exchange:{exchange},routing key:{routingKey},message:{message}");
            }
        }

        /// <summary>
        /// subscribe message
        /// </summary>
        /// <param name="queueListener">queue listener</param>
        public void Subscribe(IQueueListener queueListener)
        {
            if (_queueListener.Values.Any(a => a.QueueName == queueListener.QueueName)) throw new Exception($"{queueListener.QueueName} queue listener already exist!");

            var consumer = new EventingBasicConsumer(this._channel);
            consumer.Received += (sender, e) =>
            {
                Task.Run(async () =>
                {
                    var msg = Encoding.UTF8.GetString(e.Body);
                    try
                    {
                        if (_queueListener.TryGetValue(e.ConsumerTag, out IQueueListener listener))
                        {
                            if (listener.AutoAsk)
                                await listener.MessageReceivedAutoAskHandler(msg, e);
                            else
                            {
                                var result = await listener.MessageReceivedHandler(msg, e);

                                if (result) _channel.BasicNack(e.DeliveryTag, false, true);
                                else _channel.BasicAck(e.DeliveryTag, false);
                            }

                            if (_isLogDetail) _log.Info($"[received]{msg}");
                        }
                        else
                        {
                            _log.Fatal($"Unable to handle received message, because the handler by [DeliveryTag] was not found in listener container," +
                                       $" this message will be try requeue if it's possiable. The detail message is: \r\n" +
                                       $"DeliveryTag:{e.DeliveryTag},Exchange:{e.Exchange},RoutingKey:{e.RoutingKey},Redelivered:{e.Redelivered},ConsumerTag:{e.ConsumerTag},Message:{msg}");

                            _channel.BasicNack(e.DeliveryTag, false, true);
                        }
                    }
                    catch (Exception exception)
                    {
                        _log.Error($"Handle received message error:{msg}", exception);
                    }
                });
            };

            var cusumerTag = this._channel.BasicConsume(queueListener.QueueName, queueListener.AutoAsk, queueListener.ConsumerTag, queueListener.Arguments, consumer);
            _queueListener.AddOrUpdate(cusumerTag, queueListener, (k, v) => v);
        }
    }
}
